package com.gin.table;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Slide;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


/**
 * @author gin
 * @date 2021/4/21
 */
public class TestTableApiWindow {

    public static void main(String[] args) {

        //每隔5秒, 统计每个基站中通话成功的数量, 假设乱序
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //每隔100ms(默认值)向数据流中插入一个 Watermark
        env.getConfig().setAutoWatermarkInterval(100);
        //2、在往socket发射数据的时候 必须携带时间戳
        //根据数据时间进行处理
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        /* 连接socket获取输入的数据
        // nc -lk 8888
        // netstat -natp | grep 8888
        // 测试数据格式, 时间戳 + 数据(窗口是9000-12000, 左闭右开)
1619008072000 sid1 gin
1619008073000 sid1 soul
1619008074000 sid2 soul
1619008082000 sid2 gin
1619008083000 sid3 soul
        */
        SingleOutputStreamOperator<Tuple3<Long, String, String>> mapStream = env.socketTextStream("172.21.194.111", 8888)
                .assignTimestampsAndWatermarks(
                        //指定水印延迟3秒: Time.seconds(3)
                        //使用flink实现好的: BoundedOutOfOrdernessTimestampExtractor
                        new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {
                            //将数据中的时间字段提取出来，转成Long类型,不改变输入的数据样式
                            @Override
                            public long extractTimestamp(String line) {
                                //第一位为时间戳, 通过空格切分(使用数据源的EventTime 替换 flink的默认时间)
                                String[] split = line.split(" ");
                                return Long.parseLong(split[0]);
                            }
                        })
                .map(new RichMapFunction<String, Tuple3<Long, String, String>>() {
                    @Override
                    public Tuple3<Long, String, String> map(String value) throws Exception {
                        String[] split = value.split(" ");
                        return new Tuple3<>(Long.parseLong(split[0]), split[1], split[2]);
                    }
                });


        //创建Table API的上下文环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table table = tableEnv.fromDataStream(mapStream, "callTime.rowtime, sid, callName");
        Table result = table
                //滚动窗口
                //.window(Tumble.over("5.second").on("callTime").as("myWindow"))
                //滑动窗口
                .window(Slide.over("5.second").every("5.second").on("callTime").as("myWindow"))
                .groupBy("myWindow, sid")
                .select("sid, sid.count as sidCount, myWindow.start, myWindow.end");

        DataStream<Tuple2<Boolean, Row>> resDataStream = tableEnv.toRetractStream(result, Row.class);
        resDataStream
                // Tuple2<Boolean, Row> 中的boolean表示是否是新的计算结果,
                // false表示是计算过程中被修改的状态数据, 旧数据的值
                // true表示是计算过程中, 修改之后的结果
                .filter((FilterFunction<Tuple2<Boolean, Row>>) value -> value.f0)
                .returns(TypeInformation.of(new TypeHint<Tuple2<Boolean, Row>>() {
                }))
                .print();

        try {
            env.execute("tableApi");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
